package com.ishansong.gis.tool;

import com.mongodb.Block;
import com.mongodb.ServerAddress;
import org.bson.BsonDocument;

import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by danqiang on 2017/7/6.
 * process
 */
class MTProcessor {
    interface Processor {
        void process(Mongo newMongo, BsonDocument bsonDocument);

        String getOldDbName();

        String getNewDbName();

        String getOldCollectionName();

        String getNewCollectionName();

    }

    private Processor processor = null;
    private int threadNum = 4;

    private Mongo oldMongo = null;
    private Mongo newMongo = null;
    private final ArrayBlockingQueue<BsonDocument> queue;

    MTProcessor(int threadNum, List<ServerAddress> oldSeeds, List<ServerAddress> newSeeds) {
        this.threadNum = threadNum;
        queue = new ArrayBlockingQueue<BsonDocument>(500 * threadNum);
        oldMongo = new Mongo(oldSeeds);
        newMongo = new Mongo(newSeeds);
    }

    private void startConsumer() {
        if (processor instanceof PerfCount) {
            //
        } else if (processor instanceof PerfRead) {
            processor.process(newMongo, null);
        } else if (processor instanceof PerfWrite) {
            //
        } else {
            ExecutorService executorService = Executors.newFixedThreadPool(threadNum);
            for (int i = 0; i < threadNum; i++) {
                executorService.execute(
                        new Runnable() {
                            @Override
                            public void run() {
                                while (true) {
                                    try {
                                        BsonDocument doc = queue.take();
                                        if (doc != null) {
                                            processor.process(newMongo, doc);
                                        }
                                    } catch (InterruptedException e) {
                                        e.printStackTrace();
                                        break;
                                    } catch (Exception e) {
                                        e.printStackTrace();
                                    }
                                }

                            }
                        });
            }
        }
    }

    private void startProducer() {
        if (processor instanceof PerfCount) {
            processor.process(oldMongo, null);
        } else if (processor instanceof PerfRead) {
            //
        } else if (processor instanceof PerfWrite) {
            //
        } else {
            int threadNum = 1;
            ExecutorService executorService = Executors.newFixedThreadPool(threadNum);
            for (int i = 0; i < threadNum; i++) {
                executorService.execute(
                        new Runnable() {
                            @Override
                            public void run() {
                                oldMongo.iterateWithCollectionName(new Block<BsonDocument>() {
                                    public void apply(BsonDocument doc) {
                                        try {
                                            if (queue.isEmpty()) {
                                                System.out.println("queue is empty");
                                            }
                                            queue.put(doc);
                                        } catch (InterruptedException e) {
                                            e.printStackTrace();
                                        }

                                    }
                                }, processor.getOldCollectionName());

                            }
                        });
            }
        }
    }

    void start(Processor processor) {
        this.processor = processor;
        oldMongo.switchCollection(processor.getOldDbName(), "");
        newMongo.switchCollection(processor.getNewDbName(), processor.getNewCollectionName());

        startConsumer();
        startProducer();
    }

}
